package com.example.watermark;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.JoinedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Created with IntelliJ IDEA.
 * ClassName: TimeWindowJoin
 * Package: com.example.watermark
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-25
 * Time: 14:04
 */

//结合窗口的合流 比如两个相同的时间窗口 通过比较规定的字段 匹配出相同字段的值
public class TimeWindowJoin {
    public static void main(String[] args) throws Exception {
        //用户统计固定时间内的两条流的匹配情况 是在事件时间上 开的是事件时间的窗 就要分配时间戳和水位线 assignTAndW方法

        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        final SingleOutputStreamOperator<Tuple2<String, Integer>> tp1 = env.fromElements(
                Tuple2.of("a", 1),
                Tuple2.of("b", 2),
                Tuple2.of("c", 3)
        ).
                assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple2<String, Integer>>forMonotonousTimestamps()
                                .withTimestampAssigner((val, te) -> val.f1)
                );

        SingleOutputStreamOperator<Tuple2<String, Integer>> tp2 = env.fromElements(
                Tuple2.of("a", 12),
                Tuple2.of("e", 22),
                Tuple2.of("c", 32)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Tuple2<String, Integer>>forMonotonousTimestamps()
                        .withTimestampAssigner((val, te) -> val.f1)
        );

        //在窗口联结当中 首先调用DataStream join方法合并
        JoinedStreams<Tuple2<String, Integer>, Tuple2<String, Integer>> join = tp1.join(tp2);

        //返回的JoinedStream
        //在Joined基础出上 调用.where() equalTo()指定两条流联结的key
        DataStream<String> apply = join
                .where(value -> value.f0) //左
                .equalTo(value -> value.f0) //右
                //指定之后 开窗 滚动事件时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //待用apply传入联结窗口函数进行处理
                .apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String>() {
                    //参数分别是 第一条 第二条 流 和输出的数据类型
                    @Override
                    public String join(Tuple2 first, Tuple2 second) throws Exception {
                        return first + "<---------->" + second;
                    }
                });

        apply.print();

        env.execute();


    }
}
